package com.cictec.middleware.elasticsearch.camel.processor;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ThreadPoolExecutor;

import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.cictec.middleware.commons.utils.PMSUtils;
import com.cictec.middleware.commons.utils.PropertiesUtils;
import com.cictec.middleware.commons.utils.ThreadUtils;
import com.cictec.middleware.elasticsearch.service.ElasticsearchPositionService;

/**
 * 中间件里程计算 业务保存ES
 * @author ql
 */
@SuppressWarnings("rawtypes")
@Component
public class MiddlewareMileageCaculateSaveESProcessor implements Processor {

	private static Logger logger = LoggerFactory.getLogger(MiddlewareMileageCaculateSaveESProcessor.class);

	@Autowired
	private ElasticsearchPositionService elasticsearchPositionService;
	
	@Override
	public void process(Exchange exchange) throws Exception {

		Message message = exchange.getIn();
		Map data = (Map) message.getBody();

		startMonitory();
		
		try {
			pool.submit(new PositionSaveESRunable(data));
		} catch (Exception e) {
			e.printStackTrace();
			logger.error("【里程计算】 保存车辆位置信息到ES错误! {}", e);
		}
	}
	
	/* ******************** busMileageCaculate start *********************************************/
	/**
	 * 里程计算线程
	 * @author ql
	 *
	 */
	class PositionSaveESRunable implements Callable<String> {

		private Map data;

		public PositionSaveESRunable(Map data) {
			this.data = data;
		}

		@Override
		public String call() throws Exception {
			
			try {

				mileageCaculateHandler(data);
			} catch (Exception e) {
				e.printStackTrace();
				
				return "";
			}

			return "";
		}
	}
	
	private ThreadPoolExecutor pool = null;
	private boolean isShutdown = false;
	
	
	private void startMonitory(){
		if(pool == null){
			int threadMax = PropertiesUtils.getIntValue("busMileage.threads", 4);
			pool = ThreadUtils.getFixedThreadPool("bus-position-es-handler", threadMax);
			logger.debug("【里程计算】 开启ES存储多线程");
			
			new Thread() {
				public void run() {
					while (!isShutdown || (pool != null && pool.getQueue().size() != 0)) {
						monitor();
					}
					if(pool.isShutdown() && pool.getQueue().size() == 0){
						if(pool.getActiveCount() == 0){
							pool = null;
						}
						logger.debug("【里程计算】 关闭ES存储多线程");
					}
					
//					if (isShutdown) {
//						shutdownCaculate();
//					}
				}
			}.start();
		}
	}
	
	private void monitor(){
		logger.debug("[monitor][bus-position-es-handler][{}/{}] Active:{}, Completed:{}, Task:{}, queueSize:{}, isShutdown:{}, isTerminated:{}",
				pool.getPoolSize(), 
				pool.getCorePoolSize(), 
				pool.getActiveCount(),
				pool.getCompletedTaskCount(), 
				pool.getTaskCount(), 
				pool.getQueue().size(),
				pool.isShutdown(),
				pool.isTerminated()
				);
		int seconds = 10;
		try {
			Thread.sleep(seconds * 1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	public void shutdown(){
		try {
			// 延迟10秒关闭
			Thread.sleep(10000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		if(pool != null){
			pool.shutdown();
//			pool = null;
			isShutdown = true;
//			logger.debug("【里程计算】 关闭ES存储多线程");
		}
	}


	/**
	 * 里程计算位置信息保存至ES
	 * @param data
	 */
	private void mileageCaculateHandler(Map data) {
		String date = PMSUtils.isNull(data.get("date"));
		String busUuid = PMSUtils.isNull(data.get("busUuid"));
		String busNumber = PMSUtils.isNull(data.get("busNumber"));

		// Map map = new HashMap();
		// map.put("date", date);
		// map.put("busUuid", busUuid);
		// map.put("busNumber", busNumber);
		// map.put("deleteEnterOut", deleteEnterOutList);
		// map.put("saveEnterOut", enterOutESModelList);
		// map.put("savePosition", positionESModelList);

		String enteroutJson = PMSUtils.isNull(data.get("saveEnterOut"));
		String positionJson = PMSUtils.isNull(data.get("savePosition"));
		
		List<Map> enterOutESModelList = JSON.parseArray(enteroutJson, Map.class);
		List<Map> positionESModelList = JSON.parseArray(positionJson, Map.class);

		// 进出站信息：
		if (enterOutESModelList.size() > 0) {
			try {
				logger.debug("删除ES进出站信息：{} {}", date, busNumber);
				elasticsearchPositionService.deleteEnterOut(date, busNumber);
			} catch (Exception e) {
				e.printStackTrace();
				logger.debug("删除ES进出站信息错误：{} {} {}", date, busNumber, e);
			}

			Map<String, Object> additionalParams = new HashMap<String, Object>();
			additionalParams.put("busNumber", busNumber);
			logger.debug("保存ES进出站信息-开始：{} {}", date, busNumber);
			elasticsearchPositionService.saveEnterOutStations(enterOutESModelList, additionalParams);
			logger.debug("保存ES进出站信息-结束：{} {}", date, busNumber);
		}

		enterOutESModelList = null;
		
		
		// 位置信息保存：
		if (positionESModelList.size() > 0) {
			try {
				logger.debug("删除ES位置信息：{} {}", date, busNumber);
				elasticsearchPositionService.deletePosition(date, busUuid, busNumber);
			} catch (Exception e) {
				e.printStackTrace();
				logger.debug("删除ES位置信息错误：{} {} {}", date, busNumber, e);
			}

			
			logger.debug("保存ES位置信息开始：{} {}", date, busNumber);
			try {
				Map<String, Object> additionalParams = new HashMap<String, Object>();
				additionalParams.put("busNumber", busNumber);
				elasticsearchPositionService.savePosition(positionESModelList, additionalParams);
			} catch (Exception e) {
				e.printStackTrace();
				logger.debug("保存ES位置信息错误 ：{} {} {}", date, busNumber, e);
			}
			logger.debug("保存ES位置信息结束：{} {}", date, busNumber);
		}
		
		positionESModelList = null;
		data = null;

	}

	/* ******************** busMileageCaculate end   *********************************************/

}
